package com.example.service;

import com.example.decoder.RpcDecoder;
import com.example.handler.UserServiceHandler;
import com.example.quartz.QuartzMain;
import com.example.request.RpcRequest;
import com.example.serialization.JSONSerializer;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import org.I0Itec.zkclient.ZkClient;
import org.quartz.SchedulerException;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author lbq
 * @version 1.0
 * @date 2021/2/3 22:31
 */
public class UserServiceImpl implements UserService {

    private final ExecutorService executorService;
    private final static ZkClient zkClient = new ZkClient("127.0.0.1:2181");

    public UserServiceImpl() {
        this.executorService = Executors.newSingleThreadExecutor();
    }

    @Override
    public String sayHello(String msg) {
        System.out.println("are you ok ? " + msg);
        return "are you ok ? " + msg;
    }

    public void startServer(String ip, int port) throws InterruptedException, SchedulerException {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workGroup = new NioEventLoopGroup();

        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup, workGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        pipeline.addLast(new StringEncoder());
                        pipeline.addLast(new RpcDecoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(new UserServiceHandler());
                    }
                });

        serverBootstrap.bind(ip, port).sync();
        // 在zookeeper创建临时节点,并将服务器端口信息储存在节点上
        String server = ip + ":" + port;
        System.out.println("server = " + server);
        createNode(server);
    }

    private void createNode(String server) throws SchedulerException {
        // 创建永久节点,节点数据为服务端地址
        zkClient.createPersistent("/zkServer/server2", true);
        zkClient.writeData("/zkServer/server2", server);
        // 创建永久节点,节点数据为服务端发送心跳链接的时间戳
        zkClient.createPersistent("/zkServerHeartbeat/server2", true);
        zkClient.writeData("/zkServerHeartbeat/server2", LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli());
        QuartzMain.start("0/5 * * * * ?");
    }

    public static void updateTime() {
        zkClient.writeData("/zkServerHeartbeat/server2", LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli());
        System.out.println("更新时间节点数据" + zkClient.readData("/zkServerHeartbeat/server2"));
    }
}
